package com.atguigu.gmall.realtime.app.dwd;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/12/7 10:51
 */
public class Dwd_05_DwdTradePayDetailSuc extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_05_DwdTradePayDetailSuc().init(
            3005,
            2,
            "Dwd_05_DwdTradePayDetailSuc"
        );
        
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db
        readOdsDb(tEnv, "Dwd_05_DwdTradePayDetailSuc");
        // 2. 读取字典表
        readBaseDic(tEnv);
        // 3. 过滤出支付成功数据
        Table paymentInfo = tEnv
            .sqlQuery("select " +
                          "data['user_id'] user_id," +
                          "data['order_id'] order_id," +
                          "data['payment_type'] payment_type," +
                          "data['callback_time'] callback_time," +
                          "`et`," +
                          "`pt`," +
                          "ts " +
                          "from ods_db " +
                          "where `database`='gmall2022' " +
                          "and `table`='payment_info' " +
                          "and `type`='update' " +
                          "and `old`['payment_status'] is not null " +
                          "and `data`['payment_status']='1602'");
        tEnv.createTemporaryView("payment_info", paymentInfo);
        // 4. 读取 dwd 层的下单事务事实表
        tEnv
            .executeSql("create table dwd_trade_order_detail(" +
                            "id string," +
                            "order_id string," +
                            "user_id string," +
                            "sku_id string," +
                            "sku_name string," +
                            "province_id string," +
                            "activity_id string," +
                            "activity_rule_id string," +
                            "coupon_id string," +
                            "date_id string," +
                            "create_time string," +
                            "source_id string," +
                            "source_type string," +
                            "source_type_name string," +
                            "sku_num string," +
                            "split_original_amount string," +
                            "split_activity_amount string," +
                            "split_coupon_amount string," +
                            "split_total_amount string," +
                            "ts bigint, " +
                            "`et` as to_timestamp_ltz(ts, 0)," +
                            "watermark for `et` as `et` - interval '3' second " +
                            ")" + SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, "Dwd_04_DwdTradeOrderCancel"));
        // 5. 3 张表 join(使用 interval join)
        Table result = tEnv
            .sqlQuery("select " +
                          "od.id order_detail_id," +
                          "od.order_id," +
                          "od.user_id," +
                          "od.sku_id," +
                          "od.sku_name," +
                          "od.province_id," +
                          "od.activity_id," +
                          "od.activity_rule_id," +
                          "od.coupon_id," +
                          "pi.payment_type  payment_type_code," +
                          "dic.dic_name payment_type_name," +
                          "pi.callback_time," +
                          "od.source_id," +
                          "od.source_type source_type_code," +
                          "od.source_type_name," +
                          "od.sku_num," +
                          "od.split_original_amount," +
                          "od.split_activity_amount," +
                          "od.split_coupon_amount," +
                          "od.split_total_amount split_payment_amount," +
                          "pi.ts " +
                          "from payment_info pi " +
                          "join dwd_trade_order_detail od " +
                          "on pi.order_id=od.order_id " +
                          "   and pi.et <= od.et + interval '30' minute and pi.et >= od.et - interval '5' second " +
                          "join base_dic for system_time as of pi.pt as dic on pi.payment_type=dic.dic_code ");
        
        // 6.写出到 kafka 中
        tEnv.executeSql("create table dwd_trade_pay_detail_suc(" +
                            "order_detail_id string," +
                            "order_id string," +
                            "user_id string," +
                            "sku_id string," +
                            "sku_name string," +
                            "province_id string," +
                            "activity_id string," +
                            "activity_rule_id string," +
                            "coupon_id string," +
                            "payment_type_code string," +
                            "payment_type_name string," +
                            "callback_time string," +
                            "source_id string," +
                            "source_type_code string," +
                            "source_type_name string," +
                            "sku_num string," +
                            "split_original_amount string," +
                            "split_activity_amount string," +
                            "split_coupon_amount string," +
                            "split_payment_amount string," +
                            "ts bigint " +
                            ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_PAY_DETAIL_SUC));
        
        result.executeInsert("dwd_trade_pay_detail_suc");
        
    }
}
/*
交易域支付成功事务事实表

payment_info  update 1601->1602
下单事务事实表
    order_info    update 1001->1002
    order_detail  insert
    ...
 
字典表
 */